package drds.plus.executor.command_handler.put;

import drds.plus.executor.ExecuteContext;
import drds.plus.executor.cursor.cursor.ISortingCursor;
import drds.plus.executor.data_node_executor.DataNodeExecutorContext;
import drds.plus.executor.record_codec.RecordCodecFactory;
import drds.plus.executor.record_codec.record.Record;
import drds.plus.executor.row_values.RowValues;
import drds.plus.executor.table.ITable;
import drds.plus.executor.utils.Utils;
import drds.plus.sql_process.abstract_syntax_tree.configuration.ColumnMetaData;
import drds.plus.sql_process.abstract_syntax_tree.configuration.IndexMapping;
import drds.plus.sql_process.abstract_syntax_tree.execute_plan.dml.IPut;

import java.util.ArrayList;
import java.util.List;

public class DeleteHandler extends PutHandler {

    public DeleteHandler() {
        super();
    }

    protected int executePut(ExecuteContext executeContext, IPut put, ITable table, IndexMapping indexMapping) throws Exception {
        int affectRows = 0;
        IPut delete = put;
        ISortingCursor cursor = null;
        RowValues rowData = null;

        try {
            cursor = DataNodeExecutorContext.getExecutorContext().getDataNodeExecutor().execute(executeContext, delete.getQuery());

            List<Record> deleteRecord = new ArrayList();
            while ((rowData = cursor.next()) != null) {
                Record record = RecordCodecFactory.newRecordCodec(indexMapping.getKeyColumnMetaDataList()).newRecord();
                affectRows++;
                for (ColumnMetaData columnMetaData : indexMapping.getKeyColumnMetaDataList()) {//根据key值进行删除
                    Object object = getValueByColumnMetaData(rowData, columnMetaData);
                    record.put(columnMetaData.getColumnName(), object);
                }
                deleteRecord.add(record);
            }
            for (Record record : deleteRecord) {
                table.delete(executeContext, put.getTableName(), indexMapping, record);
            }
        } catch (Exception e) {
            throw e;
        } finally {
            if (cursor != null) {
                List<RuntimeException> exs = new ArrayList();
                exs = cursor.close(exs);
                if (!exs.isEmpty()) {
                    throw exs.get(0);
                }
            }
        }
        return affectRows;
    }

    private Object getValueByColumnMetaData(RowValues rowData, ColumnMetaData columnMetaData) {
        Object value = Utils.getValue(rowData, columnMetaData.getTableName(), columnMetaData.getColumnName(), columnMetaData.getAlias());
        return value;
    }

}
